package com.googlecode.eventio.internal;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import com.googlecode.eventio.EventDispatcherService;
import com.googlecode.eventio.Reaction;
import com.googlecode.eventio.config.Configuration;
import com.googlecode.eventio.internal.config.ConfigImpl;
import com.googlecode.eventio.internal.utils.NamedThreadFactory;
import com.googlecode.eventio.spi.FailureHandler;
import com.googlecode.eventio.spi.StatisticsImplementor;

/**
 * The {@link EventDispatcherImpl} controls event flow: it supports event
 * subscription and notification, and calls the registered reactions whenever a
 * event is triggered. The Dispatcher executes all reactions in a dedicated
 * {@link Thread}, which can be parameterized by using a custom
 * {@link ThreadFactory}.
 * 
 * @author mofleury
 * 
 */
public class EventDispatcherImpl implements EventDispatcherService {

	private final ExecutorService executor;

	private final FailureHandler failureHandler;

	private final StatisticsImplementor statisticsImplementor;

	private final Map<Object, Queue<Reaction<Object>>> registeredReactions = new HashMap<Object, Queue<Reaction<Object>>>();

	private Thread executorThread;

	public static String DEFAULT_THREAD_FACTORY_NAME = "eventio";

	public EventDispatcherImpl() {
		this(null);
	}

	public EventDispatcherImpl(Configuration configuration) {
		ConfigImpl config = new ConfigImpl();

		if (configuration != null) {
			configuration.configure(config);
		}

		if (config.getThreadFactory() == null) {
			config.threadFactory(new NamedThreadFactory(DEFAULT_THREAD_FACTORY_NAME));
		}

		if (config.getStatisticsImplementor() == null) {
			config.statistics(new NoOpStatisticsImplementor());
		}

		if (config.getFailureHandler() == null) {
			config.failureHandler(new StackTracePrintingFailureHandler());
		}

		final ThreadFactory threadFactory = config.getThreadFactory();
		failureHandler = config.getFailureHandler();
		statisticsImplementor = config.getStatisticsImplementor();

		BlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>();

		this.executor = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,//
				queue, new ThreadFactory() {

					@Override
					public Thread newThread(Runnable arg0) {
						Thread created = threadFactory.newThread(arg0);
						executorThread = created;
						return created;
					}
				});

	}

	/**
	 * Emit an event, i.e. trigger all reactions, using the provided message as
	 * reaction input
	 * 
	 * @param event
	 *            the event to trigger
	 * @param message
	 *            the message to be reacted upon
	 */
	@Override
	public <T> void emit(final Object event, final T message) {

		if (shutdown) {
			throw new RejectedExecutionException();
		}

		statisticsImplementor.emit(event);

		executor.execute(new Runnable() {
			@Override
			public void run() {

				Queue<Reaction<Object>> matchReacts = findMatches(event);

				List<Reaction<Object>> processReacts = new ArrayList<Reaction<Object>>(matchReacts);

				for (final Reaction<Object> reaction : processReacts) {
					try {
						reaction.react(message);
						statisticsImplementor.endReaction(true);
					} catch (Throwable t) {
						statisticsImplementor.endReaction(false);
						failureHandler.handleFailure(event, message, reaction, t);
					}
				}

			}
		});
	}

	/**
	 * Subscribes this reaction to the given event. The reaction will be invoked
	 * whenever the event is triggered. Subscriptions can be canceled using
	 * {@link #remove(Object, Reaction)}
	 * 
	 * @param event
	 *            the event to subscribe to
	 * @param reaction
	 *            the reaction to be executed when the event arises
	 */
	@Override
	public <T> void on(Object event, Reaction<T> reaction) {
		statisticsImplementor.on(event);
		attach(event, reaction);
	}

	@SuppressWarnings("unchecked")
	private <T> void attach(final Object event, final Reaction<T> reaction) {
		if (executorThread == Thread.currentThread()) {

			Queue<Reaction<Object>> registered = registeredReactions.get(event);
			if (registered == null) {
				registered = new LinkedList<Reaction<Object>>();
				registeredReactions.put(event, registered);
			}
			registered.offer((Reaction<Object>) reaction);
		} else {
			executor.execute(new Runnable() {

				@Override
				public void run() {
					attach(event, reaction);
				}
			});
		}
	}

	/**
	 * Same as {@link #on(Object, Reaction)}, except that the reaction will be
	 * invoked only once.
	 * 
	 * @param event
	 *            the event to subscribe to
	 * @param reaction
	 *            the reaction to be executed when the event arises
	 */
	@Override
	public <T> void once(final Object event, final Reaction<T> reaction) {
		statisticsImplementor.once(event);

		Reaction<T> selfRemovingReaction = new Reaction<T>() {
			@Override
			public void react(T input) {
				remove(event, this);
				reaction.react(input);
			};
		};

		attach(event, selfRemovingReaction);
	}

	/**
	 * Removes reaction, if any. Reaction comparison is done by reference. If
	 * the reaction is present more than once, only the most recently added
	 * instance will be removed
	 * 
	 * @param event
	 *            the event to remove reactions from
	 * @param reaction
	 *            the reaction to be removed
	 */
	@Override
	public <T> void remove(Object event, Reaction<T> reaction) {
		statisticsImplementor.remove(event);

		Queue<Reaction<Object>> registered = registeredReactions.get(event);
		if (registered != null) {
			Iterator<Reaction<Object>> reacts = registered.iterator();
			while (reacts.hasNext()) {
				if (reacts.next() == reaction) {
					reacts.remove();
					break;
				}
			}

			if (registered.isEmpty()) {
				registeredReactions.remove(event);
			}
		}
	}

	/**
	 * removes all reactions that correspond to this event
	 * 
	 * @param event
	 */
	@Override
	public void removeAll(Object event) {
		registeredReactions.remove(event);
	}

	private static Queue<Reaction<Object>> EMPTY_QUEUE = new LinkedList<Reaction<Object>>();

	private Queue<Reaction<Object>> findMatches(Object event) {
		Queue<Reaction<Object>> registered = registeredReactions.get(event);
		if (registered == null) {
			return EMPTY_QUEUE;
		}

		return registered;
	}

	/**
	 * awaits the termination of the event dispatcher executor.
	 * 
	 * @param timeOut
	 * @param timeUnit
	 * @throws InterruptedException
	 */
	@Override
	public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {

		return terminated.await(timeout, unit);

	}

	private volatile boolean shutdown = false;

	private final CountDownLatch terminated = new CountDownLatch(1);

	@Override
	public void shutdown() {
		if (!shutdown) {
			shutdown = true;

			executor.execute(new Runnable() {

				@Override
				public void run() {

					List<Runnable> remaining = executor.shutdownNow();

					/*
					 * some task may have been submitted just before this one
					 * (if it evaluated 'shutdown' before it was updated). To
					 * make sure that no tasks are silently discarded, execute
					 * any remaining tasks.
					 */
					for (Runnable r : remaining) {
						try {
							r.run();
						} catch (Throwable t) {
							// TODO notify cleanly
							t.printStackTrace();
						}
					}

					terminated.countDown();
				}
			});
		}
	}

	@Override
	public boolean isShutdown() {
		return shutdown;
	}

	@Override
	public boolean isTerminated() {
		return terminated.getCount() <= 0;
	}

}
